package com.bigdata.common.utils

import java.util.Properties

import com.bigdata.common.constants.Constants
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.util.Random

object KafkaCommonUtils {

  private val kafkaServers = ConfigurationManager.config.getString("kafka.bootstrap.servers")
  private val kafkaGroupId = ConfigurationManager.config.getString("kafka.group.id")
  private val kafkaKeySerializer = ConfigurationManager.config.getString("kafka.key.serializer")
  private val kafkaValueSerializer = ConfigurationManager.config.getString("kafka.value.serializer")

  def createKafkaProducer(topic: String): KafkaProducer[String, String] = {
    val props = new Properties()
    props.put("bootstrap.servers", kafkaServers)
    props.put("key.serializer", kafkaKeySerializer)
    props.put("value.serializer", kafkaKeySerializer)
    val producer = new KafkaProducer[String, String](props)
    producer
  }

  def createKafkaSink(env: StreamExecutionEnvironment, topic: String): DataStream[String] = {
    import org.apache.flink.api.scala._
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", kafkaServers)
    properties.setProperty("group.id", kafkaGroupId)
    properties.setProperty("key.deserializer",
      kafkaKeySerializer)
    properties.setProperty("value.deserializer",
      kafkaValueSerializer)
    properties.setProperty("auto.offset.reset", "latest")

    val source: DataStream[String] = env.addSource(new FlinkKafkaConsumer[String](topic, new SimpleStringSchema(), properties))
    source
  }

  def produceMockData(topic: String)(f: => String) = {
    val producer: KafkaProducer[String, String] = createKafkaProducer(topic)

    try {
      while (true) {
        producer.send(new ProducerRecord[String, String](topic, f))
      }
    } catch {
      case ex: Exception => ex.printStackTrace()
    }
    finally {
      producer.close()
    }
  }


}
